package top.lyunk.demo.flink.streaming;

import com.alibaba.fastjson2.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.HTable;

import java.text.SimpleDateFormat;
import java.util.Iterator;

public class App {
    public static void main(String[] args) {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // 1.获取Kafka数据
        DataStreamSource<String> source = environment.fromSource(FlinkKafkaUtils.getKafkaEventSource(), WatermarkStrategy.noWatermarks(), "Kafka Source");
        SingleOutputStreamOperator<ClientInfo> source2 = source.map(jsom -> JSON.parseObject(jsom, ClientInfo.class));

        source2.keyBy(ClientInfo::getUserId)
                .process(new KeyedProcessFunction<Long, ClientInfo, Object>() {
                    /**
                     * 从HBase读取数据
                     */
                    HTable table;

                    /**
                     * 基于状态存储用户上一次登录的数据
                     */
                    ListState<ClientInfo> lastLoginDatas;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        table = HBaseUtils.initHBaseClient();
                        lastLoginDatas = getRuntimeContext().getListState(new ListStateDescriptor<ClientInfo>("lastLoginData", ClientInfo.class));
                    }

                    @Override
                    public void processElement(ClientInfo value, KeyedProcessFunction<Long, ClientInfo, Object>.Context ctx, Collector<Object> out) throws Exception {
                        SimpleDateFormat sdf = new SimpleDateFormat();
                        String source = "0";
                        String reason = "第一次登录" + value.getAddr() + " 本次登录时间：" + value.getCreateTime();
                        try {
                            // 判断上一次登录
                            Iterator<ClientInfo> iterator = lastLoginDatas.get().iterator();
                            if (iterator.hasNext()) {
                                ClientInfo lastLoginData = iterator.next();
                                //...做操作
                                lastLoginDatas.clear();
                                lastLoginDatas.add(new ClientInfo());
                            }

                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                })
                .print();

    }
}
